package cn.itcast.dstream

import java.sql.{Connection, DriverManager, Statement}

import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.apache.spark.streaming.dstream.{DStream, ReceiverInputDStream}

object ForeachRDDTest {
    def main(args: Array[String]): Unit = {
        //1.创建SparkConf对象
        val sparkConf: SparkConf = new SparkConf().setAppName("ForeachRDDTest").setMaster("local[2]")
        //2.创建SparkContext对象,它是所有任务计算的源头
        val sc: SparkContext = new SparkContext(sparkConf)
        //3.设置日志级别
        sc.setLogLevel("WARN")
        //4.创建StreamingContext,需要2个参数，一个是SparkContext，一个是批处理的时间间隔
        val ssc: StreamingContext = new StreamingContext(sc, Seconds(5))
        //5.对接socket数据创建DStream对象，需要socket服务的地址、端口号及存储级别(默认的)
        val dstream: ReceiverInputDStream[String] = ssc.socketTextStream("192.168.121.134", 9999)
        val itemPairs: DStream[(String, Int)] = dstream.map(line => (line.split(",")(0), 1))
        val itemCount: DStream[(String, Int)] = itemPairs.reduceByKeyAndWindow((v1: Int, v2: Int) => v1 + v2, Seconds(60), Seconds(10))
        val hotWord = itemCount.transform(itemRDD => {
            val top3: Array[(String, Int)] = itemRDD.map(pair => (pair._2, pair._1)).sortByKey(false).map(pair => (pair._2, pair._1)).take(3)
            ssc.sparkContext.makeRDD(top3)
        })
        hotWord.foreachRDD(rdd => {
            val url="jdbc:mysql://192.168.121.134:3306/spark"
            val user="root"
            val password="123456"
            Class.forName("com.mysql.jdbc.Driver")
            val conn1: Connection = DriverManager.getConnection(url,user,password)
            conn1.prepareStatement("delete from searchKeyWord where 1=1").executeUpdate()
            conn1.close()
            rdd.foreachPartition(partitionOfRecords => {
                val url = "jdbc:mysql://192.168.121.134:3306/spark"
                val user = "root"
                val password = "123456"
                Class.forName("com.mysql.jdbc.Driver")
                val conn2: Connection = DriverManager.getConnection(url, user, password)
                conn2.setAutoCommit(false)
                val stat: Statement = conn2.createStatement()
                partitionOfRecords.foreach(record => {
                    stat.addBatch("insert into searchKeyWord (insert_time,keyword,search_count) values (now(),'" + record._1 + "','" + record._2 + "')")
                })
                stat.executeBatch()
                conn2.commit()
                conn2.close()
            })
            //            println(rdd.getNumPartitions)
        })
        //        hotWord.foreachRDD(rdd=>{
        //            val url="jdbc:mysql://192.168.121.134:3306/spark"
        //            val user="root"
        //            val password="123456"
        //            Class.forName("com.mysql.jdbc.Driver")
        //            val conn: Connection = DriverManager.getConnection(url,user,password)
        //            conn.prepareStatement("delete from searchKeyWord where 1=1").executeUpdate()
        //            rdd.foreachPartition(partitionOfRecords=>{
        //                val url="jdbc:mysql://192.168.121.134:3306/spark"
        //                val user="root"
        //                val password="123456"
        //                Class.forName("com.mysql.jdbc.Driver")
        //                val conn: Connection = DriverManager.getConnection(url,user,password)
        //                conn.setAutoCommit(false)
        //                val stat: Statement = conn.createStatement()
        //                partitionOfRecords.foreach(record=>{
        //                    stat.addBatch("insert into searchKeyWord (insert_time,keyword,search_count) values (now(),'"+record._1+"','"+record._2+"')")
        //                })
        //                stat.executeBatch()
        //                conn.commit()
        //
        //            })
        //            println(rdd.getNumPartitions)
        //        })
        ssc.start()
        ssc.awaitTermination()
        ssc.stop()
    }

}
